package com.stable.actor.core;

import lombok.extern.slf4j.Slf4j;

import java.io.Serializable;
import java.util.Map;
import java.util.UUID;
import java.util.concurrent.*;

/**
 * Actor引用 - 类似Akka的ActorRef
 * 适配单一端口设计
 */
@Slf4j
public class ActorRef implements Serializable {
    private static final long serialVersionUID = 1L;

    private final String path;
    private final StableActor actor;
    private final ActorSystem actorSystem;
    protected boolean isLocal;
    private String host;
    private int port = -1;

    // 用于ask模式的响应映射
    private static final Map<String, CompletableFuture<Object>> pendingResponses = new ConcurrentHashMap<>();
    private static final ScheduledExecutorService timeoutScheduler = Executors.newScheduledThreadPool(2);

    // 本地Actor引用
    public ActorRef(String path, StableActor actor, ActorSystem actorSystem) {
        this.path = path;
        this.actor = actor;
        this.actorSystem = actorSystem;
        this.isLocal = true;
        this.host = null;
        this.port = -1;

        if (actor != null) {
            actor.setSelf(this);
        }
    }

    // 远程Actor引用
    public ActorRef(String path, String host, int port, ActorSystem actorSystem) {
        this.path = path;
        this.actor = null;
        this.actorSystem = actorSystem;
        this.isLocal = false;
        this.host = host;
        this.port = port;
    }

    // 基础构造（用于Terminated消息等场景）
    public ActorRef(String path, StableActor actor, ActorSystem actorSystem, boolean isLocal) {
        this.path = path;
        this.actor = actor;
        this.actorSystem = actorSystem;
        this.isLocal = isLocal;
    }

    // 发送消息 - 类似Akka的tell方法
    public void tell(Object message, ActorRef sender) {
        if (isLocal) {
            if (actor != null) {
                actor.tell(message, sender);
            } else {
                log.warn("本地Actor为null，无法发送消息: {}", path);
            }
        } else {
            // 远程消息发送
            sendRemoteMessage(message, sender);
        }
    }

    /**
     * 发送消息（简化版，无发送者）
     */
    public void send(Object message) {
        tell(message, null);
    }

    /**
     * 询问模式 - 等待响应（类似Akka的ask）
     */
    public CompletableFuture<Object> ask(Object message, long timeoutMs) {
        String correlationId = UUID.randomUUID().toString();
        CompletableFuture<Object> future = new CompletableFuture<>();

        // 注册等待响应
        pendingResponses.put(correlationId, future);

        // 设置超时
        timeoutScheduler.schedule(() -> {
            if (pendingResponses.remove(correlationId) != null) {
                future.completeExceptionally(new TimeoutException("Ask timeout after " + timeoutMs + "ms"));
            }
        }, timeoutMs, TimeUnit.MILLISECONDS);

        if (isLocal) {
            // 本地Actor的ask
            AskWrapper wrappedMessage = new AskWrapper(message, correlationId, path);
            tell(wrappedMessage, null);
        } else {
            // 远程Actor的ask
            sendRemoteAskMessage(message, correlationId);
        }

        return future;
    }

    /**
     * 处理ask模式的响应（内部使用）
     */
    public static void handleAskResponse(String correlationId, Object response) {
        CompletableFuture<Object> future = pendingResponses.remove(correlationId);
        if (future != null) {
            future.complete(response);
        } else {
            log.warn("收到未知的ask响应，correlationId: {}", correlationId);
        }
    }

    /**
     * 转发消息 - 保持原始发送者
     */
    public void forward(Object message, ActorRef originalSender) {
        tell(message, originalSender);
    }

    /**
     * 启动Actor
     */
    public void start() {
        if (actor != null) {
            actor.start();
            log.debug("启动本地Actor: {}", path);
        }
        // 不再需要启动独立的TCP服务，使用系统统一的网络服务
    }

    /**
     * 发送远程消息
     */
    private void sendRemoteMessage(Object message, ActorRef sender) {
        try {
            if (host == null || port == -1) {
                throw new IllegalStateException("远程Actor引用缺少主机或端口信息: " + path);
            }

            String senderPath = sender != null ? sender.path() : "no-sender";

            // 使用ActorSystem的网络服务发送消息
            actorSystem.sendRemoteMessage(host, port, path, message, sender);

            log.debug("📤 发送远程消息到 {}:{} {}, 发送者: {}",
                    host, port, path, senderPath);

        } catch (Exception e) {
            log.error("❌ 发送远程消息失败: {}@{}:{}", path, host, port, e);
            // 可以在这里处理重试逻辑或死信
        }
    }

    /**
     * 发送远程ask消息
     */
    private void sendRemoteAskMessage(Object message, String correlationId) {
        try {
            if (host == null || port == -1) {
                throw new IllegalStateException("远程Actor引用缺少主机或端口信息: " + path);
            }

            // 创建ask包装消息
            RemoteAskWrapper askMessage = new RemoteAskWrapper(
                    message,
                    correlationId,
                    actorSystem.getSystemPort()  // 回复端口
            );

            // 使用ActorSystem的网络服务发送消息
            actorSystem.sendRemoteMessage(host, port, path, askMessage, null);

            log.debug("📤 发送远程ask消息到 {}:{} {}, correlationId: {}",
                    host, port, path, correlationId);

        } catch (Exception e) {
            log.error("❌ 发送远程ask消息失败: {}@{}:{}", path, host, port, e);
            // 移除等待的future
            CompletableFuture<Object> future = pendingResponses.remove(correlationId);
            if (future != null) {
                future.completeExceptionally(e);
            }
        }
    }

    /**
     * 处理远程ask响应
     */
    public void handleRemoteAskResponse(String correlationId, Object response) {
        handleAskResponse(correlationId, response);
    }

    /**
     * 停止Actor
     */
    public void stop() {
        if (actor != null) {
            actor.stop();
            log.debug("停止本地Actor: {}", path);
        }
        // 不再需要停止独立的TCP服务
    }

    public String path() {
        return path;
    }

    public boolean isLocal() {
        return isLocal;
    }

    @Override
    public String toString() {
        if (isLocal) {
            return String.format("ActorRef[local://%s (system:%s)]",
                    path, actorSystem.getName());
        } else {
            return String.format("ActorRef[remote://%s:%d/%s]", host, port, path);
        }
    }

    /**
     * 获取底层的Actor实例（仅限本地Actor）
     */
    public StableActor getActor() {
        if (!isLocal) {
            throw new UnsupportedOperationException("远程Actor引用没有本地实例: " + path);
        }
        if (actor == null) {
            throw new IllegalStateException("本地Actor引用为null: " + path);
        }
        return actor;
    }

    /**
     * 检查是否是本地Actor
     */
    public boolean hasLocalActor() {
        return isLocal && actor != null;
    }

    /**
     * 检查是否是远程Actor
     */
    public boolean isRemote() {
        return !isLocal;
    }

    /**
     * 获取主机信息（远程Actor）
     */
    public String getHost() {
        return host;
    }

    /**
     * 获取端口信息（远程Actor）
     */
    public int getPort() {
        return port;
    }

    /**
     * 获取所属的ActorSystem
     */
    public ActorSystem getActorSystem() {
        return actorSystem;
    }

    public static ActorRef noSender() {
        return null;
    }

    /**
     * Ask模式的消息包装器（本地）
     */
    private static class AskWrapper implements Serializable {
        private static final long serialVersionUID = 1L;
        private final Object originalMessage;
        private final String correlationId;
        private final String replyTo;

        public AskWrapper(Object originalMessage, String correlationId, String replyTo) {
            this.originalMessage = originalMessage;
            this.correlationId = correlationId;
            this.replyTo = replyTo;
        }

        public Object getOriginalMessage() {
            return originalMessage;
        }

        public String getCorrelationId() {
            return correlationId;
        }

        public String getReplyTo() {
            return replyTo;
        }
    }

    /**
     * 远程Ask模式的消息包装器
     */
    public static class RemoteAskWrapper implements Serializable {
        private static final long serialVersionUID = 1L;
        private final Object originalMessage;
        private final String correlationId;
        private final int replyPort;  // 回复端口

        public RemoteAskWrapper(Object originalMessage, String correlationId, int replyPort) {
            this.originalMessage = originalMessage;
            this.correlationId = correlationId;
            this.replyPort = replyPort;
        }

        public Object getOriginalMessage() {
            return originalMessage;
        }

        public String getCorrelationId() {
            return correlationId;
        }

        public int getReplyPort() {
            return replyPort;
        }
    }

    /**
     * 远程Ask响应
     */
    public static class RemoteAskResponse implements Serializable {
        private static final long serialVersionUID = 1L;
        private final String correlationId;
        private final Object response;

        public RemoteAskResponse(String correlationId, Object response) {
            this.correlationId = correlationId;
            this.response = response;
        }

        public String getCorrelationId() {
            return correlationId;
        }

        public Object getResponse() {
            return response;
        }
    }

    /**
     * 清理资源
     */
    public static void shutdown() {
        // 取消所有等待的ask请求
        for (CompletableFuture<Object> future : pendingResponses.values()) {
            future.completeExceptionally(new RuntimeException("ActorSystem shutdown"));
        }
        pendingResponses.clear();

        // 关闭超时调度器
        timeoutScheduler.shutdown();
        try {
            if (!timeoutScheduler.awaitTermination(5, TimeUnit.SECONDS)) {
                timeoutScheduler.shutdownNow();
            }
        } catch (InterruptedException e) {
            timeoutScheduler.shutdownNow();
            Thread.currentThread().interrupt();
        }
    }
}
